from typing import Dict, Any, List
import time
from intelli_port.commons.config import load_settings
from intelli_port.commons.logging import json_logger, emit_to_elastic
from intelli_port.commons.observability import provider_error_rate, histogram_p95_ms_by_label, provider_request_latency_ms, gauge_value_by_label, provider_queue_length, provider_kv_cache_hit_rate, provider_gpu_mem_used_mb
from intelli_port.model.provider_local import LocalQwenProvider
from intelli_port.model.provider_http import HttpQwenProvider
from intelli_port.model.provider_hybridflow import HybridFlowQwenProvider
from intelli_port.model.qwen.provider_vllm import VllmQwenProvider


class ProviderManager:
    def __init__(self):
        s = load_settings()
        pri = (getattr(s, "qwen_provider_priority", "local,http,hybridflow") or "local,http,hybridflow").split(",")
        self.providers: List[str] = [p.strip() for p in pri if p.strip()]
        self.local = LocalQwenProvider(s.qwen_edge_path)
        self.http = HttpQwenProvider(s.qwen_http_endpoint)
        self.hybridflow = HybridFlowQwenProvider()
        self.vllm = VllmQwenProvider(s.qwen_vllm_endpoint)
        self.edge_p95_ms = int(getattr(s, "qwen_edge_p95_ms", 300))
        self.error_rate_threshold = float(getattr(s, "qwen_error_rate_threshold", 0.01))
        self.provider_p95_max_ms = int(getattr(s, "provider_p95_max_ms", 1000))
        self.provider_queue_max = float(getattr(s, "provider_queue_max", 1.0))
        self.provider_gpu_mem_max_mb = float(getattr(s, "provider_gpu_mem_max_mb", 16000.0))
        self.provider_kv_min_hit_rate = float(getattr(s, "provider_kv_min_hit_rate", 0.1))
        self.window = 50
        self.latencies: List[int] = []
        self.errors = 0
        self.calls = 0
        self.error_streaks: Dict[str, int] = {}
        self.cooldown_until: Dict[str, float] = {}
        self.max_errors = int(getattr(s, "provider_error_streak_max", 3))
        self.cooldown_seconds = float(getattr(s, "provider_cooldown_seconds", 30.0))
        self.logger = json_logger()

    def _record(self, latency_ms: int, ok: bool):
        self.latencies.append(latency_ms)
        if len(self.latencies) > self.window:
            self.latencies.pop(0)
        self.calls += 1
        if not ok:
            self.errors += 1

    def _should_fallback(self) -> bool:
        if not self.latencies:
            return False
        p95 = sorted(self.latencies)[max(0, int(0.95 * len(self.latencies)) - 1)]
        err_rate = (self.errors / self.calls) if self.calls else 0.0
        return p95 > self.edge_p95_ms or err_rate > self.error_rate_threshold

    async def infer(self, text: str, lang: str = "zh") -> Dict[str, Any]:
        result = None
        for p in self.providers:
            if p in self.cooldown_until and time.time() < self.cooldown_until.get(p, 0):
                continue
            if self._is_unhealthy(p):
                self.cooldown_until[p] = time.time() + self.cooldown_seconds
                try:
                    self.logger.info("provider.cooldown.auto", extra={"service": "api", "provider": p})
                    emit_to_elastic({"event": "provider.cooldown.auto", "provider": p})
                except Exception:
                    pass
                continue
            try:
                if p == "local":
                    r = self.local.infer(text, lang)
                    self._record(r.get("latency_ms", 0), True)
                    result = r
                    if self._should_fallback():
                        continue
                    return r
                if p == "http":
                    r = await self.http.infer(text, lang)
                    ok = bool(r.get("reply"))
                    self._record(r.get("latency_ms", 0), ok)
                    if ok:
                        self.error_streaks[p] = 0
                    else:
                        self.error_streaks[p] = self.error_streaks.get(p, 0) + 1
                        if self.error_streaks[p] >= self.max_errors:
                            self.cooldown_until[p] = time.time() + self.cooldown_seconds
                            try:
                                self.logger.info("provider.cooldown", extra={"service": "api", "provider": p, "cooldown_seconds": self.cooldown_seconds})
                                emit_to_elastic({"event": "provider.cooldown", "provider": p, "cooldown_seconds": self.cooldown_seconds})
                            except Exception:
                                pass
                    if ok:
                        return r
                    continue
                if p == "hybridflow":
                    r = self.hybridflow.infer(text, lang)
                    ok = bool(r.get("reply"))
                    self._record(r.get("latency_ms", 0), ok)
                    if ok:
                        self.error_streaks[p] = 0
                    else:
                        self.error_streaks[p] = self.error_streaks.get(p, 0) + 1
                        if self.error_streaks[p] >= self.max_errors:
                            self.cooldown_until[p] = time.time() + self.cooldown_seconds
                            try:
                                self.logger.info("provider.cooldown", extra={"service": "api", "provider": p, "cooldown_seconds": self.cooldown_seconds})
                                emit_to_elastic({"event": "provider.cooldown", "provider": p, "cooldown_seconds": self.cooldown_seconds})
                            except Exception:
                                pass
                    if ok:
                        return r
                    continue
                if p == "vllm":
                    r = await self.vllm.infer(text, lang)
                    ok = bool(r.get("reply"))
                    self._record(r.get("latency_ms", 0), ok)
                    if ok:
                        self.error_streaks[p] = 0
                    else:
                        self.error_streaks[p] = self.error_streaks.get(p, 0) + 1
                        if self.error_streaks[p] >= self.max_errors:
                            self.cooldown_until[p] = time.time() + self.cooldown_seconds
                            try:
                                self.logger.info("provider.cooldown", extra={"service": "api", "provider": p, "cooldown_seconds": self.cooldown_seconds})
                                emit_to_elastic({"event": "provider.cooldown", "provider": p, "cooldown_seconds": self.cooldown_seconds})
                            except Exception:
                                pass
                    if ok:
                        return r
                    continue
            except Exception:
                self._record(0, False)
                self.error_streaks[p] = self.error_streaks.get(p, 0) + 1
                if self.error_streaks[p] >= self.max_errors:
                    self.cooldown_until[p] = time.time() + self.cooldown_seconds
                    try:
                        self.logger.info("provider.cooldown", extra={"service": "api", "provider": p, "cooldown_seconds": self.cooldown_seconds})
                        emit_to_elastic({"event": "provider.cooldown", "provider": p, "cooldown_seconds": self.cooldown_seconds})
                    except Exception:
                        pass
                continue
        return result or {"reply": text, "lang": lang, "provider": "fallback", "latency_ms": 0}

    def stats(self) -> Dict[str, Any]:
        if not self.latencies:
            return {"p95": 0, "error_rate": 0.0, "calls": self.calls}
        p95 = sorted(self.latencies)[max(0, int(0.95 * len(self.latencies)) - 1)]
        err_rate = (self.errors / self.calls) if self.calls else 0.0
        return {"p95": p95, "error_rate": round(err_rate, 4), "calls": self.calls}

    def set_priority(self, providers: list[str]) -> Dict[str, Any]:
        self.providers = [p.strip() for p in providers if p.strip()]
        return {"priority": self.providers}

    def provider_states(self) -> Dict[str, Any]:
        states: Dict[str, Any] = {}
        for p in self.providers:
            states[p] = {
                "cooldown_until": self.cooldown_until.get(p, 0.0),
                "error_streak": self.error_streaks.get(p, 0),
            }
        return {"states": states}

    def cooldown(self, provider: str, seconds: float | None = None) -> Dict[str, Any]:
        sec = seconds if seconds is not None else self.cooldown_seconds
        self.cooldown_until[provider] = time.time() + float(sec)
        try:
            self.logger.info("provider.cooldown.manual", extra={"service": "api", "provider": provider, "cooldown_seconds": sec})
            emit_to_elastic({"event": "provider.cooldown.manual", "provider": provider, "cooldown_seconds": sec})
        except Exception:
            pass
        return {"provider": provider, "cooldown_until": self.cooldown_until.get(provider, 0.0)}

    def _label_for(self, provider: str) -> str | None:
        if provider == "http":
            return "qwen_http"
        if provider == "vllm":
            return "qwen_vllm"
        if provider == "hybridflow":
            return "qwen_hybridflow"
        return None

    def _is_unhealthy(self, provider: str) -> bool:
        label = self._label_for(provider)
        if not label:
            return False
        try:
            err = provider_error_rate(label)
            p95 = histogram_p95_ms_by_label(provider_request_latency_ms, "name", label)
            qlen = gauge_value_by_label(provider_queue_length, "name", label)
            kv = gauge_value_by_label(provider_kv_cache_hit_rate, "name", label)
            gpum = gauge_value_by_label(provider_gpu_mem_used_mb, "name", label)
            if err > self.error_rate_threshold:
                return True
            if p95 > self.provider_p95_max_ms:
                return True
            if qlen > self.provider_queue_max:
                return True
            if gpum > self.provider_gpu_mem_max_mb:
                return True
            if kv > 0 and kv < self.provider_kv_min_hit_rate:
                return True
        except Exception:
            return False
        return False

    def resume(self, provider: str) -> Dict[str, Any]:
        self.cooldown_until.pop(provider, None)
        self.error_streaks[provider] = 0
        try:
            self.logger.info("provider.resume.manual", extra={"service": "api", "provider": provider})
            emit_to_elastic({"event": "provider.resume.manual", "provider": provider})
        except Exception:
            pass
        return {"provider": provider, "cooldown_until": self.cooldown_until.get(provider, 0.0)}